﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Messaging;
using System.Runtime.Remoting.Messaging;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using ZeroMQ;
using System.ComponentModel;

namespace Reflexive.IO
{
    public partial class ZeroReader<TConverter, TDestination> : IReflexive<TConverter, byte[], TDestination>
        where TConverter : ITypeConverter<byte[], TDestination>, new()
        where TDestination : class
    {

        IEnumerable<string> _endpoints;
        CancellationTokenSource _tokenSource;
        SynchronizedCollection<Subscription<TDestination>> _observers = new SynchronizedCollection<Subscription<TDestination>>();

        public StatusCode Status
        {
            get;
            private set;
        }

        public ZeroReader(IEnumerable<string> endpoints)
        {
            _endpoints = endpoints;
        }

        public void Stop()
        {
            try
            {
                if (!_tokenSource.IsCancellationRequested)
                {
                    _tokenSource.Cancel();
                }
            }
            finally
            {
                Status = StatusCode.Suspended;
            }
        }

        public void Start()
        {
            if (Status == StatusCode.Completed)
            {
                throw new NotSupportedException("Reader is in the Completed state.");
            }

            _tokenSource = new CancellationTokenSource();
            Task.Run(() =>
            {

                using (var ctx = ZmqContext.Create())
                {
                    using (var socket = ctx.CreateSocket(SocketType.SUB))
                    {
                        socket.SubscribeAll();
                        foreach (var endPoint in _endpoints)
                            socket.Connect(endPoint);

                        while (true)
                        {
                            try
                            {
                                int i;
                                byte[] bytes = new byte[] { };
                                var message = socket.Receive(bytes, out i);
                                try
                                {
                                    _observers.OnNext<TDestination>((Convertible<TConverter, byte[], TDestination>)message);
                                }
                                catch (Exception e)
                                {
                                    System.Diagnostics.Trace.TraceError(e.ToString());
                                }
                               
                            }
                            catch (Exception e)
                            {
                                System.Diagnostics.Trace.TraceError(e.ToString());
                                try
                                {
                                    _observers.OnError<TDestination>(e);
                                }
                                catch (Exception ex)
                                {
                                    System.Diagnostics.Trace.TraceError(ex.ToString());
                                }
                                
                            }

                        }
                    }
                }

            }, _tokenSource.Token);
            Status = StatusCode.Started;
        }

        #region Subscription Management

        public IDisposable Subscribe(IObserver<TDestination> observer)
        {
            var subscription = new Subscription<TDestination>(observer);
            subscription.UnSubscribe += UnSubscribe;
            _observers.Add(subscription);
            return subscription;
        }

        private void UnSubscribe(object sender, IObserver<TDestination> e)
        {
            _observers.Remove(((Subscription<TDestination>)sender));
        }

        #endregion

        public void Dispose()
        {
            if (Status != StatusCode.Completed)
            {
                try
                {
                    Stop();
                    _observers.OnCompleted<TDestination>();
                }
                catch (Exception ex)
                {
                    System.Diagnostics.Trace.TraceError(ex.ToString());
                }
                finally
                {
                    Status = StatusCode.Completed;
                }
            }
        }
    }
}
